KAFKA-7235: Detect outdated control requests and bounced brokers using broker generation - #5821
Conversation
|
@junrao @ijuma @lindong28 Can you help to review this PR when you have time? Thanks! |
lindong28
left a comment
lindong28
left a comment
There was a problem hiding this comment.
Thanks for the patch! Left some comments below.
lindong28
left a comment
lindong28
left a comment
There was a problem hiding this comment.
Thanks for the update! Left some more comments below.
|
@hzxa21 , @lindong28 : Yes, I will make another pass in the next couple of days. |
There was a problem hiding this comment.
Could we fix the javadoc above?
There was a problem hiding this comment.
It would be useful to avoid _1/_2 for better readability. We could do foreach { case (broker, epoch) => ...}
There was a problem hiding this comment.
It would be useful to avoid _1/_2 for better readability.
There was a problem hiding this comment.
It would be useful to avoid _1 for better readability.
There was a problem hiding this comment.
Instead of bouncedBrokerIds(broker.id), it might be clearer to do bouncedBrokerIds.contains(broker.id).
There was a problem hiding this comment.
The code in this method is quite similar to that in testControlRequestWithCorrectBrokerEpoch(). Should we merged them somehow?
There was a problem hiding this comment.
Sure. I have merged them into a single function.
There was a problem hiding this comment.
This is unnecessary. Great test!
|
@lindong28 @junrao I have updated the PR to address all the comments. Could you take a second look? |
There was a problem hiding this comment.
Could initialBrokerEpoch be right after initialBrokerInfo?
There was a problem hiding this comment.
control request => controlled shutdown request
There was a problem hiding this comment.
I mean LeaderAndIsr/UpdateMetadata/StopReplica requests here, not controlled shutdown request. I have updated the comment to avoid confusion.
There was a problem hiding this comment.
The controller part is not right. Non-controllers are registered through this api too.
There was a problem hiding this comment.
It seems that this logging is redundant since registerBroker() logs the same info already?
There was a problem hiding this comment.
Thanks for pointing out. I have removed the log here.
There was a problem hiding this comment.
Perhaps add a comment on how v1 differs from v0?
…g broker generation
|
@junrao Thanks for the prompt reply. I have updated the PR to address your comments and rebase. |
| def registerBroker(brokerInfo: BrokerInfo): Unit = { | ||
| /** | ||
| * Registers the broker in zookeeper and return the broker epoch. | ||
| * @return broker epoch (znode create transaction id) |
There was a problem hiding this comment.
Could we put return after param ?
|
|
||
| /** | ||
| * Gets all brokers with broker epoch in the cluster. | ||
| * @return sequence of brokers in the cluster. |
There was a problem hiding this comment.
Could we change the comment accordingly?
There was a problem hiding this comment.
It seems that you fixed a different line?
There was a problem hiding this comment.
Ah. My bad. I mislooked this one. Fixed.
|
@junrao Thanks for the comments. I have addressed your comments in the updated commit. |
|
@lindong28 : Do you want to take another look of this PR before merging? |
|
LGTM. Thanks for the in-depth review @junrao! I am currently not able to access my original github account due to loss of my phone number and can not merge this PR myself. |
|
@hzxa21 : Could you address the last comment on line 415 in KafkaZkClient.scala? |
|
@junrao My bad. Updated the PR to fix that. |
junrao
left a comment
junrao
left a comment
There was a problem hiding this comment.
@hzxa21 : Thanks for the latest patch. LGTM. Also note that broker epoch can potentially be used to improve KAFKA-6029 as well.
|
@junrao Thanks a lot for the review. There are several follow-ups (KAFKA-6029 & KAFKA-7283) we can do after getting the broker epoch. I will address them accordingly and submit PRs. If you are aware of anything that can also benefit from broker epoch, do let me know and I am open to more discussion. Thanks again! |
…g broker generation (apache#5821) * KAFKA-7235: Detect outdated control requests and bounced brokers using broker generation * Add broker_epoch in controlled shutdown request * Move broker epoch check into controller for ControlledShutdownRequest * Refactor schema definition for controler requests/responses * Address comments * Address comments * Address comments * Send back STALE_BROKER_EPOCH error in ControlledShutdown response * Fix build issue * Address comments * Address comments * Address comments * Address comments * Fix tests after rebase * Address comments * Address comments
4 participants
This PR introduces the broker generation concept and leverage it to allow controller to detect fast bounced brokers and allow broker to reject outdated control requests.
It has the changes required to implement KIP-380:
[Common]
multirequest directly[Broker]
[Controller]
controllerContext.brokerEpochsCacheafter reading from zk when processingBrokerChangeevent andonControllerFailoverBrokerChangeevent by comparing the broker epochs get from zk and cached broker epochs and trigger necessary state changes[Test]
BrokerEpochIntegrationTestto test broker processing new versions of the control requests and rejecting requests with stale broker epochControllerIntegrationTestto test controller detecting bounced brokersRequestResponseTestto test seralization and de-seralization for new versions of the control requestsControlRequstTestunit test to test control requests schemas normalizationCommitter Checklist (excluded from commit message)